#include "../pubsubserver.h"

using namespace pubsub;

PubSubServer::PubSubServer(EventLoop *loop, const InetAddress &listenAddr)
    : _loop(loop),
      _server(loop, listenAddr)
{
    _server.setConnectionCallback(std::bind(&PubSubServer::onConnection, this, _1));
    _server.setMessageCallback(std::bind(&PubSubServer::onMessage, this, _1, _2, _3));
    _loop->runEvery(2.0, std::bind(&PubSubServer::timePublish, this));
}

void PubSubServer::start()
{
    _server.start();
}

void PubSubServer::onConnection(const TcpConnectionPtr &conn)
{
    if (conn->connected())
    {
        conn->setContext(ConnectionSubscription());
    }
    else
    {
        const ConnectionSubscription &connSub =
            boost::any_cast<const ConnectionSubscription &>(conn->getContext());
        for (auto it = connSub.begin(); it != connSub.end();)
        {
            doUnSubscribe(conn, *it++);
        }
    }
}
void PubSubServer::onMessage(const TcpConnectionPtr &conn,
                             Buffer *buff,
                             Timestamp receiveTime)
{
    ParseResult result = kSuccess;
    while (result == kSuccess)
    {
        std::string cmd;
        std::string topic;
        std::string content;
        result = parseMessage(buff, &cmd, &topic, &content);
        if (result == kSuccess)
        {
            if (cmd == "pub")
            {
                doPublish(conn->name(), topic, content, receiveTime);
            }
            else if (cmd == "sub")
            {
                LOG_INFO << conn->name() << " subscribes " << topic;
                doSubscribe(conn, topic);
            }
            else if (cmd == "unsub")
            {
                doUnSubscribe(conn, topic);
            }
            else
            {
                conn->shutdown();
                result = kError;
            }
        }
        else if (result == kError)
        {
            conn->shutdown();
        }
    }
}
void PubSubServer::timePublish()
{
    Timestamp now = Timestamp::now();
    doPublish("internal", "utc_time", now.toFormattedString(), now);
}
void PubSubServer::doSubscribe(const TcpConnectionPtr &conn,
                               const std::string &topic)
{
    ConnectionSubscription *connSub = boost::any_cast<ConnectionSubscription>(conn->getMutableContext());

    connSub->insert(topic);
    getTopic(topic).add(conn);
}
void PubSubServer::doUnSubscribe(const TcpConnectionPtr &conn,
                                 const std::string &topic)
{
    LOG_INFO << conn->name() << " unsubscribes " << topic;
    getTopic(topic).remove(conn);
    ConnectionSubscription *connSub = boost::any_cast<ConnectionSubscription>(conn->getMutableContext());
    connSub->erase(topic);
}
void PubSubServer::doPublish(const std::string &source,
                             const std::string &topic,
                             const std::string &content,
                             Timestamp time)
{
    getTopic(topic).publish(content, time);
}

Topic &PubSubServer::getTopic(const std::string &topic)
{
    std::map<std::string, Topic>::iterator it = _topics.find(topic);
    if (it == _topics.end())
    {   
        it = _topics.insert(std::make_pair(topic, Topic(topic))).first;
    }
    return it->second;
}
